package com.cn.client;

import com.cn.client.impl.DefaultMQPushConsumerImpl;
import com.cn.common.PullRequest;
import com.cn.common.ServerThread;

import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class PullMessageService extends ServerThread {

    public final LinkedBlockingQueue<PullRequest> pullRequestTables = new LinkedBlockingQueue<>();

    private final MQClientInstance mqClientInstance;

    private ScheduledExecutorService executorService;

    public void executePullRequestImmediately(final PullRequest pullRequest) {
        try {
            System.out.println(PullMessageService.class.getName() + "->监听到变动：" + pullRequest.toString());
            this.pullRequestTables.put(pullRequest);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public PullMessageService(MQClientInstance mqClientInstance) {
        this.mqClientInstance = mqClientInstance;
        this.executorService = Executors.newSingleThreadScheduledExecutor();
    }

    @Override
    public String getServiceName() {
        return this.getClass().getSimpleName();
    }

    private void pullMessage(PullRequest pullRequest) {
        DefaultMQPushConsumerImpl defaultMQPushConsumer = this.mqClientInstance.selectDefaultMQPushConsumer(pullRequest.getConsumerGroup());
        if (defaultMQPushConsumer != null) {
            defaultMQPushConsumer.pullMsg(pullRequest);
        }
    }

    /**
     * 立刻执行拉取
     *
     * @param pullRequest
     */
    public void executePullMessageImmediately(PullRequest pullRequest) {
        this.pullMessage(pullRequest);
    }

    /**
     * 延迟指定时间拉取
     *
     * @param pullRequest
     * @param interal
     */
    public void executePullMessageLate(PullRequest pullRequest, long interal) {
        executorService.schedule(new Runnable() {
            @Override
            public void run() {
                pullMessage(pullRequest);
            }
        }, interal, TimeUnit.MILLISECONDS);
    }

    @Override
    public void run() {
        while (!this.isStopped()) {
            try {
                //如果没有新的pullrequest,那么一直阻塞
                PullRequest pullRequest = this.pullRequestTables.take();
                pullMessage(pullRequest);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    public static void main(String[] args) {

    }

}
